package cn.doitedu.eagle.ruleutils;

import cn.doitedu.eagle.beans.RuleKieSessionStateBean;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.metrics.Counter;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.internal.utils.KieHelper;

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-03-10
 * @desc 规则流数据处理工具
 */
@Slf4j
public class RuleHandler {

    /**
     * 规则输入处理入口方法
     * @return
     */
    public static boolean handleRule(BroadcastState<String, RuleKieSessionStateBean> broadcastState,
                                     String ruleId, String command,
                                     String ruleCode,
                                     String ruleEventCountSql,
                                     String ruleEventSeqSql,
                                     Counter onlineRuleSCounter) {
        try {
            switch (command) {
                case "0":
                case "1":
                    addNewRule(broadcastState,ruleId,ruleCode,ruleEventCountSql,ruleEventSeqSql);
                    onlineRuleSCounter.inc();
                    log.info("插入|更新|启用了一条规则:{} ",ruleId);
                    log.info("对应的count sql语句：{}",ruleEventCountSql);
                    log.info("对应的seq sql语句：{}",ruleEventSeqSql);

                    break;

                case "2":
                case "3":
                    deleteRule(broadcastState,ruleId);
                    onlineRuleSCounter.dec();
                    log.info("删除|停用了一条规则: {}" + ruleId);
                    break;
                default:
                    break;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        return false;
    }


    /**
     * 添加新规则
     * @throws Exception
     */
    public static String addNewRule(
            BroadcastState<String, RuleKieSessionStateBean> broadcastState,
            String ruleId,
            String ruleCode,
            String eventCountQuerySql,
            String eventSeqQuerySql) throws Exception {

        // rule代码内容放入state： (ruleId -> ruleCode)
        KieHelper helper = new KieHelper();

        // 将规则代码字符串添加到helper，并构建kieSession
        helper.addContent(ruleCode, ResourceType.DRL);
        KieSession kieSession = helper.build().newKieSession();

        // 将新的kieSession规则对象放入state
        RuleKieSessionStateBean ruleKieSessionStateBean = new RuleKieSessionStateBean(ruleId, kieSession, eventCountQuerySql, eventSeqQuerySql);
        broadcastState.put(ruleId, ruleKieSessionStateBean);
        // 返回ruleId
        return ruleId;
    }

    /**
     * 更新规则
     * @throws Exception
     */
    public static String updateRule(BroadcastState<String, RuleKieSessionStateBean> broadcastState, String ruleId, String ruleCode) throws Exception {
        // return addNewRule(broadcastState,ruleId,ruleCode);
        return  null;
    }


    /**
     * 删除规则
     * @throws Exception
     */
    public static String deleteRule(BroadcastState<String, RuleKieSessionStateBean> broadcastState, String ruleId) throws Exception {
        broadcastState.remove(ruleId);
        return ruleId;
    }

}
